package com.hxq.demo2;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/** 自定义分区器
 * @author hxq
 * @date 2022/4/15 18:01
 */
public class PartitionProducer {


    public static void main(String[] args) {

        Properties properties = setConfigInfo();
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        //发送数据
        for (int i= 1; i<= 4; i++){
            ProducerRecord<String, String> record = new ProducerRecord<>("TestTopic", "xxxxxxxxxx");
            producer.send(record,(recordMetadata,e)->{
                if (e==null){
                    System.out.println("分区："+recordMetadata.partition()+"=====" + "游标："+recordMetadata.offset());
                }else {
                    e.printStackTrace();
                }
            });
        }

        producer.close();

    }


    private static Properties setConfigInfo(){
        Properties properties = new Properties();
        //kafka集群
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"212.129.155.35:9092");
        //ack
        properties.put("ack","all");
        //重试次数
        properties.put("retries",1);
        //RecordAccumulator 缓冲区大小
        properties.put("buffer.memory", 33554432);

        properties.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        //使用自定义分区器
        properties.put("partitioner.class","com.hxq.partition.MyPartition");
        return properties;

    }
}
